package com.xxx.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;

/**
 * rocketmq消息生产者
 */
@Slf4j
@Service
public class MessageProducer {

    @Autowired
    private DefaultMQProducer defaultMQProducer;

    /**
     * 发送普通消息
     */
    public void sendMessage(String topic, String msg) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        Message message = new Message(topic, msg.getBytes());
        SendResult sendResult = defaultMQProducer.send(message);
        log.info("发送普通消息结果: " + sendResult);
    }

    /**
     * 发送带tag的消息
     */
    public void sendMessageWithTag(String topic, String tag, String msg) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        Message message = new Message(topic, msg.getBytes());
        message.setTags(tag);
        SendResult sendResult = defaultMQProducer.send(message);
        log.info("发送带tag的消息结果: " + sendResult);
    }

    /**
     * 发送异步消息
     */
    public void sendAsyncMessage(String topic, String msg) throws RemotingException, InterruptedException, MQClientException {
        Message message = new Message(topic, msg.getBytes());
        defaultMQProducer.send(message, new SendCallback() { // 发送异步消息回调
            public void onSuccess(SendResult sendResult) {
                System.out.println("发送异步成功: " + sendResult);
            }

            public void onException(Throwable e) {
                System.err.println("发送异步失败: " + e.getMessage());
            }
        });
    }

    /**
     * 发送单向消息（没有返回结果）
     */
    public void sendOneWayMessage(String topic, String msg) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        Message message = new Message(topic, msg.getBytes());
        defaultMQProducer.sendOneway(message);
        log.info("发送单向消息成功");
    }

    /**
     * 发送延时消息
     */
    public void sendDelayMessage(String topic, String msg) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        Message message = new Message(topic, msg.getBytes());
        message.setDelayTimeLevel(3);   // 设置延时消息延时级别（1~18级，4.0版本支持）
//        message.setDelayTimeSec(10);   // 设置消息延时时间：10s（需要rocketmq5.0+版本支持）
        SendResult sendResult = defaultMQProducer.send(message);
        log.info("延迟消息发送结果: " + sendResult);
    }

    /**
     * 发送顺序消息（原生实现rocketmq顺序消息消费者①）
     */
    public void sendOrderedMessage(String topic, String msg, String shardingKey) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        Message message = new Message(topic, msg.getBytes());
        SendResult sendResult = defaultMQProducer.send(message, new MessageQueueSelector() {
            public MessageQueue select(List<MessageQueue> mqs, Message message, Object arg) {  // arg就是shardingKey
                // 使用shardingKey的hash值选择队列
                int index = Math.abs(arg.hashCode()) % mqs.size();
                MessageQueue mq = mqs.get(index);
//                log.info("本地发送顺序消息使用的mq：" + mq);
                return mq;    // 相同orderId总是返回同一队列
            }
        }, shardingKey);     // orderId作为路由键
        log.info("顺序消息发送结果: " + sendResult);
    }

    /**
     * 批量发送消息
     */
    public void batchSendMessage(String topic, List<String> msgs) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        List<Message> messageList = new ArrayList<Message>();
        for (String msg : msgs) {
            Message message = new Message(topic, msg.getBytes());
            messageList.add(message);
        }

        SendResult sendResult = defaultMQProducer.send(messageList);
        log.info("批量发送消息结果: " + sendResult);
    }

}
